package com.fanxuankai.ksql.function.udf.math;

import io.confluent.ksql.function.udf.Udf;
import io.confluent.ksql.function.udf.UdfDescription;
import io.confluent.ksql.function.udf.UdfParameter;

import java.math.BigDecimal;
import java.math.RoundingMode;

/**
 * @author fanxuankai
 */
@UdfDescription(name = "divide", description = "除法计算", author = "fanxuankai")
public class Divide {
    private static final RoundingMode DEFAULT_ROUNDING_MODE = RoundingMode.HALF_UP;
    private static final String P_VAL = "val";
    private static final String P_DIVISOR = "divisor";
    private static final String P_SCALE = "scale";
    private static final String P_ROUNDING_MODE = "roundingMode";

    @Udf(description = "整数相除.")
    public Integer divide(@UdfParameter(P_VAL) Integer val, @UdfParameter(P_DIVISOR) Integer divisor) {
        if (val == null || divisor == null || divisor == 0) {
            return 0;
        }
        return val / divisor;
    }

    @Udf(description = "长整数相除.")
    public Long divide(@UdfParameter(P_VAL) Long val, @UdfParameter(P_DIVISOR) Long divisor) {
        if (val == null || divisor == null || divisor == 0) {
            return 0L;
        }
        return val / divisor;
    }

    @Udf(description = "双精度相除.")
    public Double divide(@UdfParameter(P_VAL) Double val, @UdfParameter(P_DIVISOR) Double divisor) {
        if (val == null || divisor == null || divisor == 0) {
            return 0d;
        }
        return val / divisor;
    }

    @Udf(description = "整数相除，保留小数点.")
    public Double divide(@UdfParameter(P_VAL) Integer val, @UdfParameter(P_DIVISOR) Integer divisor,
                         @UdfParameter(P_SCALE) int scale) {
        return divide(val, divisor, scale, DEFAULT_ROUNDING_MODE);
    }

    @Udf(description = "长整数相除，保留小数点.")
    public Double divide(@UdfParameter(P_VAL) Long val, @UdfParameter(P_DIVISOR) Long divisor,
                         @UdfParameter(P_SCALE) int scale) {
        return divide(val, divisor, scale, DEFAULT_ROUNDING_MODE);
    }

    @Udf(description = "双精度相除，保留小数点.")
    public Double divide(@UdfParameter(P_VAL) Double val, @UdfParameter(P_DIVISOR) Double divisor,
                         @UdfParameter(P_SCALE) int scale) {
        return divide(val, divisor, scale, DEFAULT_ROUNDING_MODE);
    }

    @Udf(description = "整数相除，保留小数点.")
    public Double divide(@UdfParameter(P_VAL) Integer val, @UdfParameter(P_DIVISOR) Integer divisor,
                         @UdfParameter(P_SCALE) int scale, @UdfParameter(P_ROUNDING_MODE) int roundingMode) {
        return divide(val, divisor, scale, RoundingMode.valueOf(roundingMode));
    }

    @Udf(description = "长整数相除，保留小数点.")
    public Double divide(@UdfParameter(P_VAL) Long val, @UdfParameter(P_DIVISOR) Long divisor,
                         @UdfParameter(P_SCALE) int scale, @UdfParameter(P_ROUNDING_MODE) int roundingMode) {
        return divide(val, divisor, scale, RoundingMode.valueOf(roundingMode));
    }

    @Udf(description = "双精度相除，保留小数点.")
    public Double divide(@UdfParameter(P_VAL) Double val, @UdfParameter(P_DIVISOR) Double divisor,
                         @UdfParameter(P_SCALE) int scale, @UdfParameter(P_ROUNDING_MODE) int roundingMode) {
        return divide(val, divisor, scale, RoundingMode.valueOf(roundingMode));
    }

    @Udf(description = "整数相除，保留小数点.")
    public Double divide(@UdfParameter(P_VAL) Integer val, @UdfParameter(P_DIVISOR) Integer divisor,
                         @UdfParameter(P_SCALE) int scale, @UdfParameter(P_ROUNDING_MODE) String roundingMode) {
        return divide(val, divisor, scale, RoundingMode.valueOf(roundingMode));
    }

    @Udf(description = "长整数相除，保留小数点.")
    public Double divide(@UdfParameter(P_VAL) Long val, @UdfParameter(P_DIVISOR) Long divisor,
                         @UdfParameter(P_SCALE) int scale, @UdfParameter(P_ROUNDING_MODE) String roundingMode) {
        return divide(val, divisor, scale, RoundingMode.valueOf(roundingMode));
    }

    @Udf(description = "双精度相除，保留小数点.")
    public Double divide(@UdfParameter(P_VAL) Double val, @UdfParameter(P_DIVISOR) Double divisor,
                         @UdfParameter(P_SCALE) int scale, @UdfParameter(P_ROUNDING_MODE) String roundingMode) {
        return divide(val, divisor, scale, RoundingMode.valueOf(roundingMode));
    }

    private <T extends Number> Double divide(T val, T divisor, int scale, RoundingMode roundingMode) {
        if (val == null || divisor == null || divisor.intValue() == 0) {
            return 0d;
        }
        return BigDecimal.valueOf(val.doubleValue())
                .divide(BigDecimal.valueOf(divisor.doubleValue()), scale, roundingMode).doubleValue();
    }
}
